package com.hzlj.position.dbsync.guangdong.listener;

import cn.hutool.core.util.ObjectUtil;
import com.fz.common.kafka.annotation.KafkaListenerRetry;
import com.hzlj.position.config.common.enums.ChannelType;
import com.hzlj.position.dbsync.guangdong.config.GuangdongLocateSyncConfig;
import com.hzlj.position.dbsync.guangdong.rest.GuangdongLocateSyncRest;
import com.hzlj.position.locate.common.common.PositionWithExtDTO;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.stereotype.Component;

import javax.annotation.Resource;
import java.util.List;
import java.util.stream.Collectors;

import static com.hzlj.position.config.common.constant.Topic.POSITION_SYNC;


/**
 * @author lifh
 * @date 2023/3/17 11:29
 **/
@Slf4j
@Component
public class GuangdongPositionSyncListener {
    @Resource
    private GuangdongLocateSyncRest guangdongLocateSyncRest;
    @Resource
    private GuangdongLocateSyncConfig guangdongLocateSyncConfig;

    @KafkaListenerRetry(enable = false)
    @KafkaListener(topics = {POSITION_SYNC})
    public void sync(ConsumerRecord<String, List<PositionWithExtDTO>> record) {
        List<PositionWithExtDTO> positions = record.value().stream()
                .filter(e -> e.getJgCode().startsWith(guangdongLocateSyncConfig.getJgCode()))
                .collect(Collectors.toList());
        log.info("【kafka】收到定位同步消息:size={}", positions.size());
        if (ObjectUtil.isEmpty(positions)) {
            return;
        }

        try {
            //运营商
            guangdongLocateSyncRest.syncOperator(positions.stream()
                    .filter(e -> e.getChannelType().equals(ChannelType.MOBILE.getCode()))
                    .filter(e -> !e.getWarning())
                    .collect(Collectors.toList())
            );
            //app
            guangdongLocateSyncRest.syncApp(positions.stream()
                    .filter(e -> e.getChannelType().equals(ChannelType.APP.getCode()))
                    .filter(e -> !e.getWarning())
                    .collect(Collectors.toList())
            );
            //错误的
            guangdongLocateSyncRest.syncWarning(positions.stream()
                    .filter(PositionWithExtDTO::getWarning)
                    .collect(Collectors.toList())
            );
        } catch (Exception e) {
            log.error("【kafka】收到定位同步消息失败:size={}", positions.size());
        }
    }


}
